package rdd.operate;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import scala.Array;
import scala.Int;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.List;

//reduceByKey是在元组，也就是kv键值对的基础上进行的
//通过kv中的k，直接将对应的v进行聚合并进行相应的计算得到一个结果值v，然后变成（k，v）
public class Spark58_Operate_reduceByKey {
    public static void main(String[] args) {
        final SparkConf conf = new SparkConf();
        //conf.setMaster("local");
        conf.setAppName("spark");
        final JavaSparkContext jsc = new JavaSparkContext(conf);

        final List<Tuple2<String,Integer>> datas = new ArrayList<>();
        datas.add(new Tuple2<String,Integer>("a",1));
        datas.add(new Tuple2<String,Integer>("a",2));
        datas.add(new Tuple2<String,Integer>("a",3));
        datas.add(new Tuple2<String,Integer>("a",4));

        final JavaRDD<Tuple2<String,Integer>> rdd = jsc.parallelize(datas);
        final JavaPairRDD<String,Integer> pairRDD = rdd.mapToPair(t -> t);
        pairRDD.collect().forEach(System.out::println);
        /*这里是不使用匿名函数的方式进行
        pairRDD.reduceByKey(
                new Function2<Integer, Integer, Integer>() {
                    @Override
                    public Integer call(Integer v1, Integer v2) throws Exception {
                        return v1 + v2 ;
                    }
                }
        ).collect().forEach(System.out::println);
        */

        //reduceByKey中定义了一个Integer类，有sum、max、min各类方法，对v进行操作
        pairRDD.reduceByKey(
                Integer::sum
        ).collect().forEach(System.out::println);

    }
}
